package org.example

import org.apache.spark.sql.SparkSession

import scala.Console.println

object Spark2 {
  def main(args: Array[String]): Unit = {
    val spark = SparkSession
      .builder()
      .master("local[*]")
      .appName("spark")
      .getOrCreate()
    val sc = spark.sparkContext
//   电影用户分析数据概况
    val filePath = "src/main/resources/"
    val moviesRDD = sc.textFile(filePath + "movies.dat")
    val occupationsRDD = sc.textFile(filePath + "occupations.dat")
    val ratingsRDD = sc.textFile(filePath + "ratings.dat")
    val usersRDD = sc.textFile(filePath + "users.dat")
    println("总电影数：" + moviesRDD.count())
    println("总用户数：" + usersRDD.count())
    println("评分条数：" + ratingsRDD.count())
    println("职业数：" + occupationsRDD.count())

    val movies2000 = moviesRDD.filter(line => line.split("::")(1).contains("(2000)"))
    val movies2000Count = movies2000.count()
    val comedyMovies2000 = movies2000.filter(line => line.split("::")(2).contains("Comedy"))
    val comedyMovies2000Count = comedyMovies2000.count()
    println(s"2000年度电影的生产数量: $movies2000Count")
    println(s"2000年度喜剧片的数量: $comedyMovies2000Count")

/*//  一、用户分析：统计某用户评过分的电影数量并打印电影详情
    val userRating = ratingsRDD.map(_.split("::"))
      .map(user => (user(1),user(0)))       //转换为key(电影ID) value(电影ID）
      .filter(_._2.equals("2688"))   //统计用户id为2688的影评数量
    println("用户id为2688的影片数为：" + userRating.count() + "：以下为电影详情 \n")

    val movieInfo = moviesRDD.map(_.split("::"))
      .map(movie => (movie(0),(movie(1),movie(2))))     //电影数据转换为 key value 型
      .join(userRating)
      .map(item => (item._1,item._2._2,item._2._1._1,item._2._1._2))
//  转换为一维的数据格式(电影id，用户id，电影名，电影类型)
    movieInfo.take(10).foreach(println)*/

// 二、电影分析：找出评分最高前十部电影和观看次数最多的十部电影
/*
val rating = ratingsRDD.map(x => x.split("::")).map {
  x => {
    (x(0), x(1), x(2))
  }
}.cache()
println("平均得分最高的前 10 名的电影名称简单版")
    rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._2._1 / x._2._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)
println("观影人数最多的前 10 部电影")
    val watchViewsInfo = rating.map(x => {
      (x._2, 1)
    }).reduceByKey((x,y) => x + y)
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
    watchViewsInfo.foreach(println(_))

    println("===================>")
    rating.map(x => (x._2, 1))
      .reduceByKey((x, y) => {
        (x + y)
      })
      .map(x => (x._2, x._1))
      .sortByKey(false)
      .take(10)
      .foreach(println)

    println("按平均分取前 10 部电影输出详情:(平均分,(movieId,Title,Genres,总分,总次 数))")
    val moviesInfo = moviesRDD.map(x => x.split("::"))
      .map(x => {
        (x(0), (x(1), x(2)))
      })
    val ratingsInfo = rating.map(x => (x._2, (x._3.toDouble, 1)))
      .reduceByKey((x, y) => {
        (x._1 + y._1, x._2 + y._2)
      })
      .map(x => (x._1, (x._2._1 / x._2._2, x._2._1, x._2._2)))
    moviesInfo.join(ratingsInfo)
      .map(info => {
        (info._2._2._1, (info._1, info._2._1._1, info._2._1._2, info._2._2._2, info._2._2._3))
      })
      .sortByKey(false)
      .take(10)
      .foreach(println)
*/




    /*val ratings = ratingsRDD.map(_.split("::"))
      .map(rating => (rating(1), (rating(2).toDouble, 1))) // (电影ID, (评分, 1))
      .reduceByKey((x, y) => (x._1 + y._1, x._2 + y._2)) // (电影ID, (总评分, 评分次数))
      .mapValues { case (totalScore, count) => totalScore / count } // (电影ID, 平均评分)
      .sortBy(_._2, ascending = false) // 按平均评分降序排序
      .take(10) // 取前10部

     println("评分最高的前十部电影：")
     ratings.foreach { case (movieId, avgScore) =>
          val movieDetails = moviesRDD.filter(line => line.split("::")(0) == movieId).first().split("::")
          println(s"电影ID: $movieId, 平均评分: $avgScore, 电影名: ${movieDetails(1)}, 类型: ${movieDetails(2)}")
        }
        // 观看次数最多的十部电影
     val watchCounts = ratingsRDD.map(_.split("::"))
          .map(rating => (rating(1), 1)) // (电影ID, 1)
          .reduceByKey(_ + _) // (电影ID, 观看次数)
          .sortBy(_._2, ascending = false) // 按观看次数降序排序
          .take(10) // 取前10部

     println("观看次数最多的十部电影：")
     watchCounts.foreach { case (movieId, count) =>
          val movieDetails = moviesRDD.filter(line => line.split("::")(0) == movieId).first().split("::")
          println(s"电影ID: $movieId, 观看次数: $count, 电影名: ${movieDetails(1)}, 类型: ${movieDetails(2)}")
        }*/
// 三、电影类型统计
/*    val movieStyle = moviesRDD.map(_.split("::"))
      .map(x => (x(0),x(2)))      //只留电影id和电影类型
      .flatMapValues(x => x.split("\\|"))      //根据符号拆分
      .map(x => (x._2,1))         //转换为 key value 类型
      .reduceByKey((x,y) => x + y)

//  在控制台打印科幻片的电影数量
    movieStyle.take(10).foreach(println)

// 四、年度电影数量统计
//  正则表达式 找出小括号里的4位数
  val pattern = Pattern.compile("(.*)(\\(\\d{4}\\))")


  val movieYear = moviesRDD.map(_.split("::"))
    .map(x => (x(1),(1)))
    .map(x => {
      var  name = ""
      var year = ""
      val matcher = pattern.matcher(x._1)
      if (matcher.find()) {
        name = matcher.group(1)
        year = matcher.group(2)
        year = year.substring(1, year.length - 1)
      }
      if (year == "") {
        (-1,1)    //标识码
      }else {
        (year.toInt,1)
      }
    })
    .reduceByKey((x,y) => x + y)
// 求2000年度电影生产数量

  movieYear.foreach(println)*/
    sc.stop()
  }

}
